package com.monkey.server.sync;

import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.monkey.constants.CommonConstants;
import com.monkey.exceptions.NodeJoinException;
import com.monkey.model.Address;
import com.monkey.model.Node;
import com.monkey.netty.client.NettyClient;
import com.monkey.netty.client.RpcRequest;
import com.monkey.netty.server.RpcResponse;
import com.monkey.server.sync.ParentSynchronizer;

import lombok.extern.slf4j.Slf4j;

/**
 * a new ${NODE} will use method "join" to join cluster
 * and the aim node use method "accept" to save the ${NODE}
 * @author jurimengs
 *
 * @param <T>
 */
@Slf4j
public class NodeSynchronizer extends ParentSynchronizer implements InitializingBean {
    private Node self;
    
    /**
     * 
     * @Description: join an exist node, and the aim node will synchronize the new one to other nodes
     * @param address
     * @return all exist nodes
     * @author zhouman
     * @date 2020-08-04 02:21:14
     */
    public List<Node> join(Address address) {
        log.info("NodeSynchronizer.join address: {}", JSON.toJSONString(address));
        // 1. 把自己发送到目标  address， 如果发送失败则无限重试直到成功
        return joinNode(address, self);
    }

    /**
     * 已有节点接受新的节点
     * 1. 本地缓存
     * 2. 同步到其他节点
     * 3. 返回所有节点给请求加入的节点
     * @Title: accept
     * @param newNode
     * @return
     * @author zhouman
     * @date 2020-08-10 02:47:19
     */
    public List<Node> accept(Node newNode) {
        log.info("NodeSynchronizer.accept node: {}", JSON.toJSONString(newNode));
        List<Node> toNodes = ClusterContext.getNodes();
        if(toNodes.contains(newNode)) {
            log.error("existed nodecode : " + newNode.getNodeCode());
            throw new NodeJoinException("existed save node code : " + newNode.getNodeCode());
        }
        // 1. 更新本地列表
        synchronized (lock) {
            if(!toNodes.contains(newNode)) {
                toNodes.add(newNode);
            }
        }
        // 2. 拿到本地列表，并把 node 信息异步推给其他所有的节点,失败的一直重试， 重试大于 N 次后发出警告, 并持久化到本地
        syncToAllNodes(toNodes, newNode);
        return toNodes;
    }
    
    public void syncToAllNodes(List<Node> nodes, Node newNode) {
        // 1. 把自己发送到目标  address， 如果发送失败则无限重试直到成功
        // 2. 如果是自己则不需要发请求同步
        for (Node otherNode : nodes) {
            // 发通知前排除掉新节点 和 自己
            if(!newNode.equals(otherNode) && !self.equals(otherNode)) {
                log.info("sycn new Node {} to other node {} ", 
                        JSON.toJSONString(otherNode.getAddress()), JSON.toJSONString(newNode));
                // 理论上来讲， 所有的目标返回的集合应该与本地 tmp 完全相同, 可以考虑将所有的结果拿回来做对比， 以防消息丢失
                List<Node> listAfterSync = joinNode(otherNode.getAddress(), newNode);
                // 记录下目标同步的结果，这样就可以定时匹配多节点是否保持一致了
                sychResult.put(otherNode, listAfterSync);
            }
        }
    }
    
    @Override
    public void afterPropertiesSet() throws Exception {
        self = getLocalNode();
        // another thread to print host cpu
        ScheduledExecutorService printer = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread th = new Thread(r);
                th.setName(PRINTER);
                return th;
            }
        });
        printer.scheduleAtFixedRate(() -> {
            // TODO 
            log.info("TODO 扫描是否有需要同步的节点");
            // reSynchronized nodes from sychResult...
        }, initialDelay, period, TimeUnit.SECONDS);
    }
    
    private List<Node> joinNode(Address address, Node self) {
        RpcRequest request = new RpcRequest();
        request.setId(UUID.randomUUID().toString());
        request.setArgs(new Object[] {self});
        request.setMethod("accept");
        request.setBeanName(CommonConstants.NODE_SYNCHRONIZER);
        request.setParameterTypes(new Class<?>[] { Node.class });
        
        log.info("address : {}", JSON.toJSONString(address));
        NettyClient nettyClient = ClusterContext.getNettyClient(address);
        RpcResponse res = nettyClient.writeAndFlush(request);
        if(res.isSuccess()) {
            // 2. 成功后拿到对方返回的 address 集合
            @SuppressWarnings("unchecked")
            List<Node> list = (List<Node>) res.getData();
            // 3. 本地存储起来。
            ClusterContext.setNodes(list);
            return list;
        }
        log.error("加入集群失败, {}", res.getMsg());
        throw new NodeJoinException("加入集群失败: " + res.getMsg());
    }
    
    private static final String PRINTER = "NodeSyncThread";
    private static final int period = 5;
    private static final int initialDelay = 0;
    
    private Map<Node, List<Node>> sychResult = new HashMap<>();
    private byte[] lock = new byte[1];
}
